package com.young.rabbitmq.retry;

import com.rabbitmq.client.*;
import com.young.rabbitmq.util.ConnectionUtil;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 模拟消费失败，然后转发消息到失败重试队列并设置过期时间，在过期时间结束后由重试队列将自动将消息转投到其所绑定的死信交换机（
 * 该死信交换机为正常队列交换机，在生产者代码中可以体现）
 *
 * 若重试超过最大次数，则将消息投递到死信队列
 */
public class MyConsumer {
    // 最大重试次数
    static final int MAX_RETRY_COUNT=3;
    public static void main(String[] args) throws IOException {
        Connection connection = ConnectionUtil.buildConnection();
        Channel channel = connection.createChannel();
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    System.out.println("接收到消息："+new String(body,"utf-8"));
                    System.out.println("==================================================");
                    // 模拟业务出错
                    int a= 1/0;
                } catch (Exception e) {
                    System.out.println("进入失败重试..........");
                    // 获取消息属性的头部信息，并判断是否包含重试次数属性
                    Map<String, Object> headers = properties.getHeaders();
                    if (headers!=null && headers.containsKey("retryCount")){
                        // 获取当前的重试次数，并判断是否超过最大重试次数
                        Integer retryCount = (Integer) headers.get("retryCount");
                        if (retryCount<MAX_RETRY_COUNT){
                            // 未超过最大重试次数，将消息设置过期时间，并重新发送到重试队列，在消息过期后会自动以转发到死信队列的方式转到正常队列，让正常队列再次消费
                            System.out.println("重试次数："+(retryCount+1));
                            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties().builder();
                            builder.expiration("30000");
                            Map<String,Object> map = new HashMap<>();
                            map.put("retryCount",retryCount+1);
                            AMQP.BasicProperties basicProperties = builder.headers(map).build();
                            channel.basicPublish("retry_exchange","",basicProperties,body);
                        }else{
                            // 超过最大重试次数，转到死信队列
                            System.out.println("超过最大重试次数，不在重试，进入死信队列");
                            channel.basicPublish("dlx_exchange","",null,body);
                        }
                    }

                }finally {
                    // 最后一定要ACK,不然消息会一直会同时存在原始队列和重试队列中，这样就会出现一些问题
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };

        channel.basicConsume("fail_queue",false,consumer);

    }
}
